package main

import (
    "os"
    "my_git_repos/gointro/pipeline"
    "bufio"
    "fmt"
    "strconv"
)

func main() {
    p := CreateNetWorkPipeline("large.in", 8000000, 8)
    //CreateNetWorkPipeline("small.in", 512, 4)
    //time.Sleep(time.Hour)
    WriteToFile(p, "large.out")
    printFile("large.out")
}
func printFile(s string) {
    file, err := os.Open(s)
    if err != nil {
        panic(err)
    }
    defer file.Close()

    p := pipeline.ReaderSource(file, -1)
    count := 0
    for v := range p {
        fmt.Println(v)
        count++
        if count == 99 {
            break
        }
    }
}
func WriteToFile(p <-chan int, s string) {
    file, err := os.Create(s)
    if err != nil {
        panic(err)
    }
    defer file.Close()

    writer := bufio.NewWriter(file)
    defer writer.Flush()
    pipeline.WriteSink(writer, p)
}

func CreatePipeline(filename string, fileSize, chunkCount int) <-chan int {
    chunkSize := fileSize / chunkCount
    sortResult := []<-chan int{}
    pipeline.Init()
    for i := 0; i < chunkCount; i++ {
        file, err := os.Open(filename)
        if err != nil {
            panic(err)
        }

        file.Seek(int64(i * chunkSize), 0)

        source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)

        sortResult = append(sortResult, pipeline.InMemSort(source))
    }
    return pipeline.MergeN(sortResult...)
}

func CreateNetWorkPipeline(filename string, fileSize, chunkCount int) <-chan int {
    chunkSize := fileSize / chunkCount
    sortAddr := []string{}
    pipeline.Init()
    for i := 0; i < chunkCount; i++ {
        file, err := os.Open(filename)
        if err != nil {
            panic(err)
        }

        file.Seek(int64(i * chunkSize), 0)

        source := pipeline.ReaderSource(bufio.NewReader(file), chunkSize)

        addr := ":" + strconv.Itoa(7000+i)
        pipeline.NetWorkSink(addr, pipeline.InMemSort(source))
        sortAddr = append(sortAddr, addr)
    }
    //return nil
    sortResults := []<-chan int {}
    for _, addr := range sortAddr {
        sortResults = append(sortResults, pipeline.NetWorkSource(addr))
    }
    return pipeline.MergeN(sortResults...)
}